#include <iostream>
#include <pthread.h>
#include <map>
#include <cnode.h>
#include <stdint.h>

#include <getopt.h>
#include <stdarg.h>
#include <assert.h>
#include <stdlib.h>
#include <math.h>
#include <exception>

#include "MyService.h"
#include <MyService_server.skeleton.h>
#include <transport/TSocket.h>
#include <transport/TBufferTransports.h>
#include <protocol/TBinaryProtocol.h>


#include <vector>
#include <vector>
#include <string>
#include <sstream>
#include <iomanip>

using namespace apache::thrift;
using namespace apache::thrift::protocol;
using namespace apache::thrift::transport;

// the name space specified in the thrift file
using namespace mp2;




//include client and server .h files
using namespace std;
/**
 * Local function used when a remote findSuccessor is called
 * reqID : the checked id
 * return the successor of the reqID
 */
cmyID myNode::local_find_successor(cmyID reqID)
{
	cmyID np = local_find_predecessor(reqID);
	return rpCall(np.port, cmyID(-1,-1), GETSUCCESS);
}

/**
 * Called when find predecssor is called remotely or by findSuccessor
 * Returns the predecessor of reqID
 */
cmyID myNode::local_find_predecessor(cmyID reqID)
{
	cmyID np(mine), old;
	/**/
	cmyID mp = rpCall(np.port, cmyID(-1,-1), GETSUCCESS);
	while(!np.checkRangeEq(mp, reqID, m))
	{
			old = np;
			np = rpCall(np.port, reqID, CLOSESTPREC);
			//cout << "GOT CLOSEST np=" << np.id << endl; 
			if(np == old)
			{
				//cout << "BREAKING BECAUSE np=" << np.id << " and old=" << old.id << endl;
				break;
			}

		mp = rpCall(np.port, cmyID(-1,-1), GETSUCCESS);
		//cout << "My SUCCESSOR IS: " << mp.id << endl;
		if(mp.id==0){
		 	//cout << "BREAKING BECAUSE mp.id=0" << endl;
			break;
		}	
	}
	//cout << "DONE with FIND: " << np.id;
	//cout<<"Before sending"<<np.port<<" "<<np.id<<endl;	
	return np;
}
	/**
 * Finds the closet preceding finger for reqID
 */
cmyID myNode::local_closest_preceding_finger(cmyID reqID)
{ 
	//cout << "PRECEDING FINGER of: " << reqID.id << endl;
	for (int i=m; i>=1;i--)
	{
		cmyID temp = (finger_table.find(i)->second).node;
		if(mine.checkRange(reqID,temp, m))
		{
			//cout << "FOUND FINGER: " << temp.id << endl;
			return temp;	
		}
	}
	return mine;
}

/**
 * Called when a node updates its predecessor
 * The function compares the file IDs with the new predecessor
 *
 * Input: New predecessor
 * Deletes the file locally and called a remote add file to transfer
 *
 */
void myNode::transfer_files(cmyID node)
{
//	cout << "Transfering Files " << endl;
			pthread_rwlock_wrlock(&lock);
	for(map<int,file>::iterator it = files.begin(); it != files.end(); it++)
	{
		if((it->first) <= node.id)
		{
	//		cout << "transfering: " << it->first << endl;
			rpCallString(node.port, it->first, it->second.data, it->second.fname, ADDFILE);
				files.erase(it);
		}	
	} 
			pthread_rwlock_unlock(&lock);
}

// checks if incoming node
// 	can replace the current predecessor
// 	if so, predecessor is updated
void myNode::local_notify(cmyID node)
{
		// check if node's key is between the predecessor and my own
		//cout<<"NotifyLocalDATA"<<predecessor.id<<" "<<mine.id<<" "<<node.id<<endl;
		if(predecessor.isNULL() || predecessor.checkRange(mine, node, m))
		{
				if(node.id != mine.id){
						// if so, update the predecessor
						transfer_files(node);
						predecessor = node;
						cout << "node= " << mine.id << ": updated predecessor= " << predecessor.id << endl;
				}
		}
		if(mine.id==0 && node.id!=mine.id)
		{						
			predecessor = node;
		}
}

// joins the chord group by
// 	notifying the introducer 
void myNode::join(cmyID introducer)
{
	predecessor.setID(-1,-1);
	if(mine.id == 0)
		successor = mine;
	else
		successor = rpCall(introducer.port, mine, FINDSUCCESS);
	cout << "node= " << mine.id << ": initial successor= " << successor.port << endl;
}
/**
 * Find file searches for the node which would contain the key
 */
cmyID myNode::findFile(int key)
{
		cmyID node = local_find_successor(cmyID(key,0));
		return node;
}

string get_ADD_FILE_result_as_string(string fname,int key,int nodeId)
{
        std::stringstream s;
        s << "fname= " << fname << "\n";
        s << "key= " << key << "\n";
        s << "added to node= " << nodeId << "\n";
        return s.str();
}


// adds file to the correct node
// The functions checks if the find should be added to the node itself
// Else it would search for the needed successor
// And then it would call add file remotely
string myNode::local_add_file(int key, string data, string fname)
{
	cmyID node;
	// check if I need to search for owner
	int end = mine.id==0 ? pow(2,m) : mine.id;
	if(key>predecessor.id && key<=end)
	{
		node = mine;
	}
	else 
	{
		// check if I need to search
		node =findFile(key);
	}
	string ret = "";
	if(node == mine) // add File to map
	{	
		pthread_rwlock_wrlock(&lock);
			//int index = files.size();
			files[key].fname = fname;
			files[key].data = data;
		pthread_rwlock_unlock(&lock);
		cout << "node= " << mine.id <<": added file k= "
			<< key << endl;
		 ret = get_ADD_FILE_result_as_string(fname, key,mine.id);

	}
	else // rp call another node
	{
		ret = rpCallString(node.port, key, data,fname, ADDFILE);
	}
	return ret;
}

string get_DEL_FILE_result_as_string(string fname, int key, bool deleted, int nodeId)
    {
        std::stringstream s;
        s << "fname= " << fname << "\n";
        s << "key= " << key << "\n";
        if (deleted) {
            // then nodeId is meaningful
            s << "was stored at node= " << nodeId << "\n";
            s << "deleted\n";
        }
        else {
            // assume that this means file was not found
            s << "file not found\n";
        }
        return s.str();
    }


// Called when a remote del file is called
// The input is the key and the file name
// The process is the same as the adding files
string myNode::local_del_file(int key, string fname)
{
	cmyID node;
	// check if I need to search for owner
	int end = mine.id==0 ? pow(2,m) : mine.id;
	if(key>predecessor.id && key<=end)
	{
		node = mine;
	}
	else 
	{
		// check if I need to search
		node =findFile(key);
	}

	string ret ="";
	// check if I need to search
	//cmyID node =findFile(key);
	map<int,file>::iterator it;
	if(node == mine)
	{
		pthread_rwlock_wrlock(&lock);
			bool deleted = false;
			it = files.find(key);
			if(it == files.end())
			{
				cout << "node= " << mine.id <<": no such file k= "
					<< key << " to delete" << endl;
			}
			else
			{
				deleted = true;
				cout << "node= " << mine.id <<">: deleted file k= "
					<< key << endl;
				files.erase(it);
			}
		ret = get_DEL_FILE_result_as_string(fname,key,deleted,mine.id);
		pthread_rwlock_unlock(&lock);
	}
	else
	{
		ret = rpCallString(node.port, key, "",fname, DELFILE);
	}
	return ret;
}


string get_GET_FILE_result_as_string( string fname, int key, bool found, int nodeId, string fdata)
    {
        std::stringstream s;
        s << "fname= " << fname << "\n";
        s << "key= " << key << "\n";
        if (found) {
            // then nodeId is meaningful
            s << "stored at node= " << nodeId << "\n";
            s << "fdata= " << fdata << "\n";
        }
        else {
            // assume that this means file was not found
            s << "file not found\n";
        }
        return s.str();
    }

// Called for remote get file
// Same process as delete and add files
string myNode::local_get_file(int key, string fname)
{
	string ret;
	map<int,file>::iterator it;
	cmyID node;
	// check if I need to search for owner
	int end = mine.id==0 ? pow(2,m) : mine.id;
	if(key>predecessor.id && key<=end)
	{
		node = mine;
	}
	else 
	{
		// check if I need to search
		node =findFile(key);
	}

	// check if I need to search
	if(node == mine)
	{	
		pthread_rwlock_rdlock(&lock);
			it = files.find(key);
			bool found = true;
			if(it == files.end())
			{
				found = false;
				ret = "";
				cout << "node= " << mine.id <<": no such file k= "
					<< key << " to serve" << endl;
			}
			else
			{
				ret = it->second.data;
				cout << "node= " << mine.id <<": served file: k= " 
				<< key << endl;
			}
		ret = get_GET_FILE_result_as_string(fname,key, found, mine.id, ret);
		pthread_rwlock_unlock(&lock);
	}
	else
	{
		ret = rpCallString(node.port, key, "",fname, GETFILE);
	}
	return ret;
}


string get_finger_table_as_string(map<int,finger> &table, int m, int myid, int idx_of_entry1)
{
    std::stringstream s;
	s << "finger table:\n";
	for(int i = 1; i <= m; i++)
		{
		using std::setw;
        	s << "entry: i= " << setw(2) << i << ", interval=["
	          << setw(4) << (myid + (int)pow(2, i-1)) % ((int)pow(2, m))
        	  << ",   "
	          << setw(4) << (myid + (int)pow(2, i)) % ((int)pow(2, m))
        	  << "),   node= "
	          << setw(4) << table[i].node.id
        	  << "\n";

		}
    return s.str();
}

string get_keys_table_as_string(std::map<int, file>& table)
{
    std::stringstream s;
    std::map<int, file>::const_iterator it = table.begin();
    /* std::map keeps the keys sorted, so our iteration will be in
     * ascending order of the keys
     */
    s << "keys table:\n";
    for (; it != table.end(); ++it) {
        using std::setw;
        /* assuming file names are <= 10 chars long */
        s << "entry: k= " << setw(4) << it->first
          << ",  fname= " << setw(10) << it->second.fname
          << ",  fdata= " << it->second.data
          << "\n";
    }
    return s.str();
}


string get_GET_TABLE_result_as_string(map<int,finger>& finger_table,int m,int myid,int idx_of_entry1, std::map<int, file>& keys_table)
    {
        return get_finger_table_as_string(
            finger_table, m, myid, idx_of_entry1) \
            + \
            get_keys_table_as_string(keys_table);
    }

// Called when an RPC get table
// We follow the same process to process files
string myNode::local_get_table(int id)
{
	string ret ="";
	stringstream str;
	cmyID nodeSuccess, node;
	if(id == mine.id)
	{
		ret = get_GET_TABLE_result_as_string(finger_table,m,mine.id,0, files);
	}
	else
	{
		// get the predecessor of the successor
		node = local_find_successor(cmyID(id,0));
		return rpCallString(node.port, id, "","", GETTABLE);
	}
	//cout<<"Find table is over"<<endl;
	return ret;
}

/**
 * Generic call for making RPC calls for related file processes
 * Input is the call to make (choice)
 * Output is the string
 */
string myNode::rpCallString(uint32_t port, int id_key, string data, string fname, int choice)
{
	/* these next three lines are standard */
	boost::shared_ptr<TSocket> socket(new TSocket("localhost", port));
	boost::shared_ptr<TTransport> transport(new TBufferedTransport(socket));
	boost::shared_ptr<TProtocol> protocol(new TBinaryProtocol(transport));
	/* prepare arguments for the rpc call */

	/* I am a MyServiceClient */
	MyServiceClient client(protocol);
	transport->open();
	string result = "";
	/* make the call and get the return value */
	switch(choice)
	{
		case ADDFILE:
			client.addFile(result, id_key, data, fname);
			break;
		case DELFILE:
			client.delFile(result, id_key, fname);
			break;
		case GETFILE:
			client.getFile(result, id_key, fname);
			break;
		case GETTABLE:
			client.getTable(result, id_key);
			break;
		default:
			break;
	}
	transport->close();
	return result;
}

/**
 * Generic function for RP CALLS
 * The choice is the type of calls
 * Return is the cmyID
 */
cmyID myNode::rpCall(uint32_t port, cmyID arg, int choice)
{

	/* these next three lines are standard */
	boost::shared_ptr<TSocket> socket(new TSocket("localhost", port));
	boost::shared_ptr<TTransport> transport(new TBufferedTransport(socket));
	boost::shared_ptr<TProtocol> protocol(new TBinaryProtocol(transport));
	/* prepare arguments for the rpc call */
	/*MyObject input;*/
	myID input;
	input.id = arg.id;
	input.port = arg.port;

	/* I am a MyServiceClient */
	MyServiceClient client(protocol);
	transport->open();
	myID result;
	/* make the call and get the return value */
	switch(choice)
	{
		case 0:
				client.notify(input);
				break;
		case 1:
				client.getPredecessor(result, input);
				break;
		case 2:
				client.findSuccessor(result, input);
				break;
		case 3:
				client.getSuccessor(result);
				break;
		case 4:
				client.closestPrecedingFinger(result, input);
				break;
	}
	transport->close();
	if(choice)
	{
			cmyID retVal(result.id, result.port);
			return retVal;
	}	else
	{
			return cmyID(-1,-1);
	}
}	


void * fix_fingers(void * n)
{
		myNode * node = (myNode *) n;
	while(1)
	{
			//cout << "FIXING FINGERS: " << endl;
			//node->local_get_table();
			int i = rand() % node->m +1;
			cmyID old = node->finger_table[i].node;
			node->finger_table[i].node = node->local_find_successor(cmyID(node->finger_table[i].start,0));
			if(old.id != node->finger_table[i].node.id)
				cout << "node= " << node->mine.id << 
				": updated finger entry i=" << i << ", pointer= "
				<< node->finger_table[i].node.id << endl;
			sleep(node->fixInterval);
	}
	return 0;
}
		
// called every so often to make sure the successor
// 	has not changed
void * stabilize(void * n)
{
	myNode * node = (myNode *) n;
	sleep(node->stabilizeInterval);

	while(1)
	{
		if(!node->successor.isNULL())
		{
			cmyID x = node->rpCall(node->successor.port, node->successor, GETPRED);
			// check if mine < x < successor
			if(node->mine.checkRange(node->successor, x, node->m))
			{
				node->successor = x;
				//cout << "node= " << node->mine.id << ": updated successor= " << node->successor.id << endl;
			}
			if(node->mine==node->successor && !x.isNULL())
			{
				node->successor=x;
				//cout << "node= " << node->mine.id << ": updated successor= " << node->successor.port << endl;
			}
			node->rpCall(node->successor.port, node->mine, NOTIFY);
		}
		sleep(node->stabilizeInterval);
	}
	return 0;
}
void donothing(const char *) {}


int main(int argc, char **argv)
{

apache::thrift::GlobalOutput.setOutputFunction(donothing);

//Create datastructures
int opt;
    int long_index;
    int m = -1;
    int id = -1;
    int port = -1;
    int introducerPort = -1;
    int stabilizeInterval = -1;
    int fixInterval = -1;
    int seed = -1;
    const char *logconffile = NULL;

    struct option long_options[] = {
        /* mandatory args */
        {"m", required_argument, 0, 1000},
        /* id of this node: 0 for introducer */
        {"id", required_argument, 0, 1001},
        /* port THIS node will listen on, at least for the
         * Chord-related API/service
         */
        {"port", required_argument, 0, 1002},
        /* optional args */
        /* if not introducer (id != 0), then this is required: port
         * the introducer is listening on.
         */
        {"introducerPort", required_argument, 0, 1003},
        /* path to the log configuration file */
        {"logConf", required_argument, 0, 1004},
        /* intervals (seconds) for runs of the stabilization and
         * fixfinger algorithms */
        {"stabilizeInterval", required_argument, 0, 1005},
        {"fixInterval", required_argument, 0, 1006},
        {"seed", required_argument, 0, 1007},
        {0, 0, 0, 0},
    };

    while ((opt = getopt_long(argc, argv, "", long_options, &long_index)) != -1)
    {
        switch (opt) {
        case 0:
            if (long_options[long_index].flag != 0) {
                break;
            }
            printf("option %s ", long_options[long_index].name);
            if (optarg) {
                printf("with arg %s\n", optarg);
            }
            printf("\n");
            break;
        case 1000:
            m = strtol(optarg, NULL, 10);
            assert((m >= 3) && (m <= 10));
            break;
        case 1001:
            id = strtol(optarg, NULL, 10);
            assert(id >= 0);
            break;
        case 1002:
            port = strtol(optarg, NULL, 10);
            assert(port > 0);
            break;
        case 1003:
            introducerPort = strtol(optarg, NULL, 10);
            assert(introducerPort > 0);
            break;
        case 1004:
            logconffile = optarg;
            break;
        case 1005:
            stabilizeInterval = strtol(optarg, NULL, 10);
            assert(stabilizeInterval > 0);
            break;
        case 1006:
            fixInterval = strtol(optarg, NULL, 10);
            assert(fixInterval > 0);
            break;
        case 1007:
            seed = strtol(optarg, NULL, 10);
            break;
        default:
            exit(1);
        }
    }

    /* if you want to use the log4cxx, uncomment this */
    // configureLogging(logconffile);
    assert((m >= 3) && (m <= 10));
    assert((id >= 0) && (id < pow(2,m)));
    assert (port > 0);
    srand(seed);

//	usleep(rand()%10+1);
myNode * node = new myNode(m,id,port,cmyID(0,introducerPort));
node->stabilizeInterval = stabilizeInterval;
node->fixInterval = 10 /*fixInterval*/;
//Start client thread
//Start server thread 
pthread_t server_thread, stabilize_t, fix_t;
struct arg_struct * data = new struct arg_struct;
data->port=port;
data->node = node;
pthread_create(&server_thread, NULL, startServer, (void *)data);


while(1)
{
	int c = 0;
	try
	{
		node-> rpCall(node -> mine.port, cmyID(-1,-1), GETSUCCESS);
	}
	catch (exception& e)
	{
		c=1;
	}
	if(c==0) break;
	usleep(100);
}

node->join(node->introducer);
pthread_create(&stabilize_t, NULL, stabilize, (void *) node);
pthread_create(&fix_t, NULL, fix_fingers, (void *) node);


while(1)
{
;
}


return 0;
}
